/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.icanci.loopstack.damc.core.server;

import java.util.concurrent.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.google.common.collect.Maps;

import cn.hutool.http.Method;
import cn.hutool.json.JSONUtil;
import cn.icanci.loopstack.damc.api.http.Client;
import cn.icanci.loopstack.damc.api.http.HttpClientImpl;
import cn.icanci.loopstack.damc.api.http.RemoteException;
import cn.icanci.loopstack.damc.cache.DamcCenterConfigRepository;
import cn.icanci.loopstack.damc.core.server.utils.IPUtils;
import cn.icanci.loopstack.damc.spi.model.R;
import cn.icanci.loopstack.damc.spi.model.UriConstant;
import cn.icanci.loopstack.damc.spi.model.pubsub.RegisterDTO;

/**
 * @author icanci
 * @since 1.0 Created in 2022/11/20 21:29
 */
@Service
public class RegisterServer {
    private static final Logger             logger         = LoggerFactory.getLogger(RegisterServer.class);
    /** http实例 */
    private static final Client             CLIENT         = HttpClientImpl.getInstance();

    private static final int                CORE_SIZE      = Runtime.getRuntime().availableProcessors();

    private static final ThreadPoolExecutor REGISTER_POOL  = new ThreadPoolExecutor(CORE_SIZE,             //
        CORE_SIZE << 1,                                                                                    //
        60L,                                                                                               //
        TimeUnit.SECONDS,                                                                                  //
        new LinkedBlockingQueue<>(2000),                                                                   //
        runnable -> new Thread(runnable, "RegisterClient Biz Pool-" + runnable.hashCode()),                //
        (r, executor) -> {
            throw new RuntimeException("RegisterClient Biz Pool is EXHAUSTED!");
        });

    /** 注册请求地址 */
    private static final String             REQ_URL_FORMAT = "http://%s:%s%s";

    /**
     * 将SDK所在服务注册到注册中心
     *
     * @param serverAddress 服务端ip地址
     * @param serverPort 服务端端口
     * @param appCode 客户端服务code
     */
    public void register(String serverAddress, int serverPort, int clientPort, String appCode) {
        // 从Agent缓存获取注册中心地址
        // 从Agent缓存获取当前被Mock服务需要被注册的端口号
        String[] addresses = serverAddress.split(";");

        // 注册地址打散，防止压力放在同一个机器上
        for (String address : addresses) {
            try {
                FutureTask<R> task = new FutureTask<>(new RegisterCallable(address, clientPort, appCode, serverPort));
                REGISTER_POOL.execute(task);
                R r = task.get(10, TimeUnit.SECONDS);
                if (r.isOk()) {
                    break;
                } else {
                    logger.error("Task Register Exception:{}", r.getMessage());
                }
            } catch (RemoteException | ExecutionException | InterruptedException | TimeoutException e) {
                logger.warn("Register Exception:{}", e.getMessage());
            }
        }
    }

    /**
     * 启动本地Netty服务
     */
    public void startNettyServer() {
        // 注入注册bean
        DamcNettyServer.setRegisterService(this);
        String serverAddress = DamcCenterConfigRepository.getDamcCenterAddress();
        int serverPort = DamcCenterConfigRepository.getDamcCenterPort();
        int clientPort = DamcCenterConfigRepository.getMockedProjectPort();
        String appCode = DamcCenterConfigRepository.getMockedProjectCode();

        // 启动本地netty服务器
        DamcNettyServer.startClient(serverAddress, serverPort, clientPort, appCode);
    }

    /** 注册器  */
    private static class RegisterCallable implements Callable<R> {

        private final String address;
        private final int    clientPort;
        private final String appCode;
        private final int    serverPort;

        public RegisterCallable(String address, int clientPort, String appCode, int serverPort) {
            this.address = address;
            this.clientPort = clientPort;
            this.appCode = appCode;
            this.serverPort = serverPort;
        }

        @Override
        public R call() throws Exception {
            RegisterDTO registerDTO = new RegisterDTO(IPUtils.getHostIpAddress(), clientPort, appCode);
            String reqUrl = String.format(REQ_URL_FORMAT, address, serverPort, UriConstant.REGISTER);

            Client.RpcRequest rpcRequest = new Client.RpcRequest(reqUrl, registerDTO, Maps.newHashMap(), Method.POST, 3, TimeUnit.SECONDS, 3);

            R call = CLIENT.call(rpcRequest, R.class);

            logger.info("[RegisterCallable][call] Register result:{}", JSONUtil.toJsonStr(call));

            return call;
        }
    }
}
